package com.chen.base.module.redis.config;

import cn.hutool.core.map.MapUtil;
import com.chen.base.module.redis.annotation.RedisConsumer;
import com.chen.base.utils.AnnotationRouteUtil;
import com.chen.base.utils.SpringContextHolder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;

import java.lang.reflect.Method;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

/**
 * @Author: 陈锦龙
 * @Date: 2025/7/3 14:13
 */
@Slf4j
@Configuration
public class RedisDataConsumerConfig implements ApplicationRunner {

    @Override
    public void run(ApplicationArguments args) throws Exception {
        Map<Object, Map<Method, RedisConsumer>> beansMap = SpringContextHolder.getBeansByMethodAnnotate(RedisConsumer.class);
        if (MapUtil.isEmpty(beansMap)){
            return;
        }
        // 唯一性判断
        AnnotationRouteUtil.uniqueAnnotation(beansMap, RedisConsumer::value, "redis数据消费者重复");

        for (Map.Entry<Object, Map<Method, RedisConsumer>> beanEntry : beansMap.entrySet()) {
            Map<Method, RedisConsumer> methodMap = beanEntry.getValue();
            if (MapUtil.isEmpty(methodMap)){
                continue;
            }

            Object bean = beanEntry.getKey();
            for (Map.Entry<Method, RedisConsumer> methodEntry : methodMap.entrySet()) {
                Method method = methodEntry.getKey();
                RedisConsumer consumer = methodEntry.getValue();
                log.info("初始化redis队列消费者: {}", consumer.value());

                // 异步死循环阻塞方式持续监听队列中的数据
                RedisTemplate<String, Object> redisTemplate = SpringContextHolder.getBean(consumer.redisTemplateBeanName());
                CompletableFuture.runAsync(() -> {
                    String sourceKey = consumer.value();
                    String historyKey = sourceKey + consumer.historyKey();
                    long timeout = consumer.timeout();
                    TimeUnit unit = consumer.unit();
                    int historyCustomerCount = consumer.historyCustomerCount();

                    while (true){
                        try {
                            // 以阻塞的方式从队列中获取值
                            Object value = redisTemplate.opsForList().rightPopAndLeftPush(sourceKey, historyKey, timeout, unit);
                            if (value != null){
                                // 构建方法入参，如果第一个参数类型等于params类型，则传递实参进去
                                int count = method.getParameterCount();
                                Object[] metArgs = new Object[count];
                                if (count >= 1) {
                                    metArgs[0] = value;
                                }
                                // 执行主要数据消费逻辑
                                method.invoke(bean, metArgs);
                                // 保留历史消费数
                                historyCustomer(redisTemplate, historyKey, historyCustomerCount);
                            }
                        }catch (Exception e){
                            log.error("{} 队列消费异常: {}", this.getClass().getSimpleName(), e);
                        }
                    }
                });
            }
        }
    }

    /**
     * 保留多少条历史消费的记录
     */
    public void historyCustomer(RedisTemplate<String, Object> redisTemplate, String historyKey, int count){
        Long size = redisTemplate.opsForList().size(historyKey);
        if (size != null && size > count){
            redisTemplate.opsForList().rightPop(historyKey);
        }
    }

}
